package trident.state;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.LocalDRPC;
import org.apache.storm.trident.TridentState;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.builtin.Count;
import org.apache.storm.trident.operation.builtin.FilterNull;
import org.apache.storm.trident.operation.builtin.MapGet;
import org.apache.storm.trident.testing.FixedBatchSpout;
import org.apache.storm.trident.testing.MemoryMapState;
import org.apache.storm.trident.testing.Split;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class TridentWordCount {
    private static Logger loggerFactory = LoggerFactory.getLogger(TridentWordCount.class);

    public static void main(String[] args) throws Exception {
        FixedBatchSpout spout = new FixedBatchSpout(
                new Fields("sentence"),
                3,
                new Values("hello world"), new Values("hello ")
        );

        spout.setCycle(true);

        TridentTopology tridentTopology = new TridentTopology();

        TridentState tridentState = tridentTopology.newStream("spout1", spout)
                /* split by blank */
                .each(new Fields("sentence"), new Split(), new Fields("word"))
                /* Word Count */
                .groupBy(new Fields("word")).persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"));

        LocalDRPC localDRPC = new LocalDRPC();
        /* function 调用时的函数名称 */
        tridentTopology.newDRPCStream("words", localDRPC)
                .each(new Fields("args"), new Split(), new Fields("word"))
                .groupBy(new Fields("word")).stateQuery(tridentState, new Fields("word"), new MapGet(), new Fields("count"))
                /* filter lines with count is null */
                .each(new Fields("count"), new FilterNull())
                .project(new Fields("word", "count"));

        Config config = new Config();
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("wordCounter", config, tridentTopology.build());

        for (int i = 0; i < 10; i++) {
            loggerFactory.info("DRPC RESULT: {}", localDRPC.execute("words", "hello world dog"));
            Thread.sleep(1000);
        }
    }
}
